/*
 * Singularity Engine
 * Copyright (C) 2010 Vyacheslav Vorobyov, All rights reserved.
 * See LICENSE for full license information.
 */
package org.singularity.io;

import java.io.IOException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;


/**
 * @author vjache
 * 
 */
public class QueuedBlockingBridge<T> implements Input<T>, Output<T> {

	private final LinkedBlockingQueue<DME<T>> _queue;
	private AtomicBoolean _inputClosed = new AtomicBoolean(false);

	/**
	 * 
	 */
	public QueuedBlockingBridge(int aCapacity) {
		_queue = new LinkedBlockingQueue<DME<T>>(aCapacity);
	}

	/**
	 * 
	 */
	public QueuedBlockingBridge(LinkedBlockingQueue<DME<T>> aQueue) {
		_queue = aQueue;
	}

	@Override
	public CloseableIterator<DME<T>> iterator() {
		return new CloseableIterator<DME<T>>() {

			@Override
			public void close() throws IOException {
				_inputClosed.set(true);
			}

			@Override
			public boolean hasNext() {
				return !_queue.isEmpty() || !_inputClosed.get();
			}

			@Override
			public DME<T> next() {
				while (hasNext()) {
					try {
						DME<T> item = _queue.poll(100, TimeUnit.MILLISECONDS);
						if (item != null)
							return item;
					} catch (InterruptedException e) {
						throw new RuntimeException(e);
					}
				}
				return null;
			}

			@Override
			public void remove() {}
		};
	}

	@Override
	public CloseablePipe<DME<T>> open() {
		return new CloseablePipe<DME<T>>() {

			@Override
			public void close() throws IOException {
				//TODO: Handle bridge closing :_inputClosed.set(true);
			}

			@Override
			public void flush() throws IOException {
				while (!_queue.isEmpty())
					try {
						Thread.sleep(100);
					} catch (InterruptedException e) {
						throw new IOException(e);
					}
			}

			@Override
			public void push(DME<T> aItem) throws IOException {
				if (_inputClosed.get())
					throw new IOException("Closed state.");

				try {
					_queue.put(aItem);
				} catch (InterruptedException e) {
					throw new IOException(e);
				}
			}

		};
	}

}
